package main

import (
	"DMSKafka/config"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"log"
)

// 消费者
func main() {
	//获取kafka参数
	conf, err := config.GetConfig()
	if err != nil {
		log.Printf("fail to start consumer, err:%+v\n", err)
		return
	}
	//初始化消费者
	consumer, err := sarama.NewConsumer(conf.Host, nil)
	if err != nil {
		log.Printf("%+v\n", err)
		return
	}
	// 根据topic取到所有的分区
	pList, err := consumer.Partitions(conf.Topic)
	if err != nil {
		log.Printf("%+v\n", errors.Wrap(err, "获取分区失败"))
		return
	}
	fmt.Println(pList)
	//根据分区创建消费者
	for partition := range pList {
		csu, err := consumer.ConsumePartition(conf.Topic, int32(partition), sarama.OffsetNewest)
		if err != nil {
			log.Printf("partition %d 消费者创建失败,err:%+v\n", partition, err)
			return
		}
		defer csu.AsyncClose()
		// 异步从每个分区消费信息
		go func(sarama.PartitionConsumer) {
			for msg := range csu.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
			}
		}(csu)
	}
	//阻塞进程
	select {}
}
